package abacab.remote.oio;

import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.util.LinkedList;
import java.util.Queue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.remote.Connection;

/**
 * The thread in charge of writing into a socket
 */
public final class SocketWriter implements Runnable {
	/** The logger */
	private static final Logger	logger	= LoggerFactory.getLogger(SocketWriter.class);

	/** The connection */
	private final Connection		connection;

	/** The output stream */
	private final OutputStream	outputStream;

	/** The sending queue */
	private final Queue<byte[]>	sendingQueue;

	/** Indicates if the writer is stopped */
	private volatile boolean		stopped;

	/**
	 * Create the writer
	 * 
	 * @param connection
	 *          the connection
	 * @param socket
	 *          the socket
	 * @throws IOException
	 *           if an error occurred
	 */
	public SocketWriter(Connection connection, Socket socket) throws IOException {
		this.connection = connection;
		this.outputStream = socket.getOutputStream();
		this.sendingQueue = new LinkedList<byte[]>();
		Thread t = new Thread(this, "socket writer: " + socket.getRemoteSocketAddress());
		t.setDaemon(true);
		t.start();
	}

	/**
	 * Stop the writer
	 */
	public void stop() {
		this.stopped = true;
	}

	/**
	 * Send a message on the connection
	 * 
	 * @param message
	 *          the message
	 */
	public void send(byte[] message) {
		synchronized (this.sendingQueue) {
			boolean empty = this.sendingQueue.isEmpty();
			this.sendingQueue.offer(message);
			if (empty) {
				this.sendingQueue.notify();
			}
		}
	}

	@Override
	public void run() {
		try {
			while (!this.stopped) {
				byte[] buffer;
				synchronized (this.sendingQueue) {
					do {
						buffer = this.sendingQueue.poll();
						if (buffer != null) {
							break;
						}
						else if (this.stopped) {
							return;
						}
						this.sendingQueue.wait(5000L);
					} while (true);
				}
				try {
					this.outputStream.write(buffer.length >> 8);
					this.outputStream.write(buffer.length);
					this.outputStream.write(buffer);
				}
				catch (IOException e) {
					if (!this.stopped) {
						logger.error("Failed to write to the remote system", e);
						this.connection.close();
					}
				}
			}
		}
		catch (InterruptedException e) {
			logger.warn("The socket writer was interrupted: stopping the connection");
			this.connection.close();
		}
	}
}